package com.ruoyi.web.controller.mq;

import com.ruoyi.common.core.domain.AjaxResult;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

/**
 * rocketMq Controller测试接口
 */
@RestController
@RequestMapping("/rocketMq")
public class RocketMqController {

    // 订阅主题
    @GetMapping("/consumer")
    public AjaxResult subscriber() {
        try {
            // 声明并初始化一个consumer
            // 需要一个consumer group名字作为构造方法的参数，这里为consumer-gas
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-gas");
            consumer.setInstanceName("gas-instance");
            // consumer.setVipChannelEnabled(false);
            // NameServer地址列表，多个nameServer地址用分号隔开
            consumer.setNamesrvAddr("127.0.0.1:9876");

            // 这里设置的是一个consumer的消费策略
            // CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
            // CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            // CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            // 设置consumer所订阅的Topic和Tag，*代表全部的Tag
            consumer.subscribe("gas", "info");

            // 设置一个Listener，主要进行消息的逻辑处理
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList,
                                                                ConsumeConcurrentlyContext context) {

                    for (MessageExt messageExt : msgList) {
                        try {
                            String body = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                            System.out.println(Thread.currentThread().getName() +  " 获得的消息: " + body);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    // 返回消费状态
                    // CONSUME_SUCCESS 消费成功
                    // RECONSUME_LATER 消费失败，需要稍后重新消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 调用start()方法启动consumer
            consumer.start();
            System.out.println("RocketMq消费者启动成功...");

        } catch (Exception e) {
            e.printStackTrace();
        }
        // 返回
        return AjaxResult.success();
    }
}
